package edu.purdue.cs.aggr;

import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import edu.purdue.cs.aggr.AccumulatorFactory.AccumulatorType;
import edu.purdue.cs.aggr.accumulators.MicrobenchAccumulator.GrowthFactor;
import edu.purdue.cs.aggr.message.AccumulatorMessage;
import edu.purdue.cs.aggr.message.EventNotificationMessage;
import edu.purdue.cs.aggr.message.EventNotificationMessage.Event;
import edu.purdue.cs.aggr.message.EventMessageBuilder;

public class Vertex extends Node implements Externalizable {

	private List<String> sourceVertices = new ArrayList<String>();
	private String destinationVertex = null;
	@SuppressWarnings("rawtypes")
	private Map<String, Accumulator> accumulatorsFromChildren = new HashMap<String, Accumulator>();
	boolean leafVertex = false;
	String hdfsNamenode = null;
	String hdfsFilename = null;
	boolean setup = false;
	AccumulatorType accumulatorType = null;
	Integer accumulatorSubtype = 0;
	private boolean doneProcessing = false;
	private Accumulator storedAccumulator;
	private VertexManager manager;
	private int expectedUpdates = 0;
  private int receivedUpdates = 0;
  private Set<String> beginningUpdateReceivedFrom = new HashSet<String>();
	
  public Vertex() {
    super(null, 0, false);
  }
  
	public Vertex(String IP, int port, VertexManager manager) {
		super(IP, port, false);
		this.manager = manager;
	}
	
	public boolean isSetup() {
		return setup;
	}

	public void setSetup(boolean setup) {
		this.setup = setup;
	}

	public boolean isLeafVertex() {
		return leafVertex;
	}

	public void setLeafVertex(boolean leafVertex) {
		this.leafVertex = leafVertex;
	}

	public String getHdfsNamenode() {
		return hdfsNamenode;
	}

	public void setHdfsNamenode(String hdfsNamenode) {
		this.hdfsNamenode = hdfsNamenode;
	}

	public String getHdfsFilename() {
		return hdfsFilename;
	}

	public void setHdfsFilename(String hdfsFilename) {
		this.hdfsFilename = hdfsFilename;
	}

	public AccumulatorType getAccumulatorType() {
		return accumulatorType;
	}

	public void setAccumulatorType(AccumulatorType accumulatorType) {
		this.accumulatorType = accumulatorType;
	}

	public Integer getAccumulatorSubtype() {
		return accumulatorSubtype;
	}

	public void setAccumulatorSubtype(Integer accumulatorSubtype) {
		this.accumulatorSubtype = accumulatorSubtype;
	}
	
	public List<String> getSourceVertices() {
		return sourceVertices;
	}
	
	public final VertexManager getManager()
  {
    return manager;
  }

  public final void setManager(VertexManager manager)
  {
    this.manager = manager;
  }

  public void addAccumulator(Accumulator acc) throws AggrException {
		synchronized (accumulatorsFromChildren) {
      // if we already have all the accumulators, this must be an update
		  
      if (accumulatorsFromChildren.size() == sourceVertices.size()) receivedUpdates++;

		  accumulatorsFromChildren.put(acc.getSourceId(), acc);

		  System.out.println(this.getId() + ": adding acc, list size: " + accumulatorsFromChildren.size() + " receivedupdates: " + receivedUpdates + " expectedupdates: " + expectedUpdates);
		  
	//		System.out.println("accumulatorsFromChildren.size():" + accumulatorsFromChildren.size() + "  sourceVertices.size():" + sourceVertices.size());
			if (accumulatorsFromChildren.size() == sourceVertices.size() && expectedUpdates == receivedUpdates) {
				setup = true; // Accumulators are ready to be processed.
//				synchronized (this) {
//				  notifyAll();
//				}
				
				processAccumulators();
				
	//			System.out.println("notifyAll called...");
			}
		}
	}
	
	public void setSourceNodes(List<String> sourceVertices) {
		this.sourceVertices = sourceVertices;
	}
	
	public String getDestinationVertex() {
		return destinationVertex;
	}
	
	public void setDestinationVertex(String destinationVertex) {
		this.destinationVertex = destinationVertex;
	}
	
	public boolean processAccumulators() throws AggrException {
	  System.out.println(this.getId() + ": processAccumulators(). accumulatorsFromChildren.size()=" + accumulatorsFromChildren.size() + ", sourceVertices.size()=" + sourceVertices.size());
		synchronized (accumulatorsFromChildren) {
			if (accumulatorsFromChildren.size() == sourceVertices.size()) {
				long start = System.currentTimeMillis();
				Accumulator accumulator = combineAccumulators();
				// Send accumulator to the destination node
				if (destinationVertex != null) {
				  accumulator.setSourceId(this.getId());
				  System.out.println(this.getId() + ": Sending accumulator to destination. source=" + accumulator.getSourceId());
					sendAccumulatorToDestination(accumulator);
				} else {
					manager.sendMessage(EventMessageBuilder.createEventMessage(manager, Event.ROOT_DONE, null));
					System.out.println("ROOT FINISHED...");
				}
				long end = System.currentTimeMillis();
	//			System.out.println("Aggregation processing time :" + (end-start));
				
				System.out.println("Vertex done...");
				return true;
			}
		}
		return false;
	}
	
	@SuppressWarnings({ "rawtypes", "unchecked" })
	private Accumulator combineAccumulators() {
		synchronized (accumulatorsFromChildren) {
			List<Accumulator> accumulatorsFromChildren2 = new ArrayList<Accumulator>(accumulatorsFromChildren.values());
			Accumulator accumulator = accumulatorsFromChildren2.remove(0);
			accumulator.addAccumulators(accumulatorsFromChildren2);
			return accumulator;
		}
	}

	private void sendAccumulatorToDestination(Accumulator accumulator) {
		AccumulatorMessage accMessage = new AccumulatorMessage();
		accMessage.setAccumulator(accumulator);
		accMessage.setDestination(destinationVertex);
		manager.sendMessage(accMessage);
	}

	@Override
	public void run() {
		
//		System.out.println("Starting vertex ... ");
		
		while (!setup) {
			try {
				//Wait until setup
				synchronized (this) {
					wait(); //TODO remove this wait for non leaves. Just wait for incoming messages
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
		// Perform the assigned work and exit, but not if we are on a vertex that isn't our original
		// location
		if (leafVertex && manager.getId().equals(getId())) {
			try {
//				System.out.println("Leaf vertex...");
				EventNotificationMessage doneProcessing = new InputReader().readInputAndBuildAccumulator(/*hdfsNamenode, hdfsFilename,*/ accumulatorType, accumulatorSubtype, this);
				manager.sendMessage(doneProcessing);
				//TODO(kbarlow) don't exit, wait for update messages forever 
				
			} catch (AggrException e) {
				e.printStackTrace();
			}
		} else {
		}
		doneProcessing = true;
//		System.out.println("Finished vertex ... ");
	}
	
	
	//TODO call this when a message is received from controller.
	//TODO send above message
	public void sendStoredAccumulatorToDestination() throws AggrException {
		if (storedAccumulator == null) {
			throw new AggrException("Stored accumulator is null");
		}
		if (destinationVertex == null) {
			throw new AggrException("destinationVertex is null");
		}
		sendAccumulatorToDestination(storedAccumulator);
	}
	
	public boolean isDoneProcessing () {
		return doneProcessing;
	}
	
	public Accumulator getStoredAccumulator() {
		return storedAccumulator;
	}

	public void setStoredAccumulator(Accumulator storedAccumulator) {
		this.storedAccumulator = storedAccumulator;
	}
	
	 @Override
	  public void writeExternal(ObjectOutput out) throws IOException {
	    super.writeExternal(out);
	    out.writeObject(sourceVertices);
	    out.writeObject(destinationVertex);
      out.writeObject(accumulatorsFromChildren);
      out.writeObject(leafVertex);
      out.writeObject(hdfsNamenode);
      out.writeObject(destinationVertex);
      out.writeObject(hdfsFilename);
      out.writeObject(setup);
      out.writeObject(accumulatorType);
      out.writeObject(accumulatorSubtype);
      out.writeObject(doneProcessing);
      out.writeObject(storedAccumulator);
      out.writeObject(expectedUpdates);
      out.writeObject(receivedUpdates);
      out.writeObject(beginningUpdateReceivedFrom);
	  }

	  @Override
	  public void readExternal(ObjectInput in) throws IOException,
	      ClassNotFoundException {
	    super.readExternal(in);
      sourceVertices = (List<String>) in.readObject();
      destinationVertex = (String) in.readObject();
      accumulatorsFromChildren = (Map<String, Accumulator>) in.readObject();
      leafVertex = (Boolean) in.readObject();
      hdfsNamenode = (String) in.readObject();
      destinationVertex = (String) in.readObject();
      hdfsFilename = (String) in.readObject();
      setup = (Boolean) in.readObject();
      accumulatorType = (AccumulatorType) in.readObject();
      accumulatorSubtype = (Integer) in.readObject();
      doneProcessing = (Boolean) in.readObject();
      storedAccumulator = (Accumulator) in.readObject();
      expectedUpdates = (Integer) in.readObject();
      receivedUpdates = (Integer) in.readObject();
      beginningUpdateReceivedFrom =  (Set<String>) in.readObject();
	  }

    public void incrementExpectedUpdates(String sender)
    {
      if(!beginningUpdateReceivedFrom.contains(sender))
        expectedUpdates++;
      beginningUpdateReceivedFrom.add(sender);
    }
    
    public String toString() {
      return getId() + ","
          + storedAccumulator;
    }
}
